package mysql.socket;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.io.IOException;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;


/**
 * @ClassName NettyCachePool
 * @Description TODO
 * @Author fengk
 * @Date 2018/9/4 16:56
 */

public class NettyCachePool {
    private static EventLoopGroup group=new NioEventLoopGroup();
    private static Bootstrap boot=new Bootstrap();
    private static Map<Channel,SocketChannel> chManager=new ConcurrentHashMap<>();
    private static final Logger logger= LoggerFactory.getLogger(NettyCachePool.class);
    static{
        boot.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .option(ChannelOption.SO_KEEPALIVE,true)
                .option(ChannelOption.SO_REUSEADDR,true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addLast(new BusinessHandler());
                    }
                });
    }

    public static SocketChannel open(SocketAddress address) throws Exception{
        SocketChannel socket=null;
        ChannelFuture future=boot.connect(address).sync();
        if(future.isSuccess()){
            future.channel().pipeline().get(BusinessHandler.class).latch.await();
            socket=chManager.get(future.channel());
        }
        if(null==socket){
            throw  new IOException("Can not create socket!");
        }
        return socket;
    }
    public static class BusinessHandler extends SimpleChannelInboundHandler<ByteBuf> {

        private NettyCacheSocketChannel   socket = null;
        private final CountDownLatch latch  = new CountDownLatch(1);

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            socket.close();
            //socket.setChannel(null);
            chManager.remove(ctx.channel());// 移除
            super.channelInactive(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            socket = new NettyCacheSocketChannel();
            socket.setChannel(ctx.channel());
            chManager.put(ctx.channel(), socket);
            latch.countDown();
            super.channelActive(ctx);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            if (socket != null) {
                socket.writeCache(msg);
            } else {
                // TODO: need graceful error handler.
                logger.error("no socket available.");
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            // need output error for troubeshooting.
            logger.error("business error.", cause);
            ctx.close();
        }
    }
}
